package com.shujia.mllib

import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

object Demo5ImaegRead {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[6]")
      .appName("image")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
      * 读取图片数据
      *
      */


    val imageData: DataFrame = spark.read
      .format("image")
      .load("D:\\课件\\机器学习数据\\手写数字\\train")


    val nameAndFeatures: DataFrame = imageData
      .select($"image.origin" as "origin", $"image.data" as "data")
      .map {
        case Row(origin: String, data: Array[Byte]) =>

          //将数据进行标准化，黑像素点使用0代替，白的像素点使用1代替
          val point: Array[Double] = data.map(i => {
            if (i.toInt >= 0) {
              0.0
            } else {
              1.0
            }
          })

          //将特征转换成向量
          val features: linalg.Vector = Vectors.dense(point)


          //获取图片名
          val name: String = origin.split("/").last

          (name, features)
      }
      .toDF("name", "features")


    //读取图片对应的目标值
    val labels: DataFrame = spark.read
      .format("csv")
      .option("sep", " ")
      .schema("name STRING, label DOUBLE")
      .load("D:\\课件\\机器学习数据\\手写数字\\train.txt")

    //关联标签，取出标签和特征
    val data: DataFrame = nameAndFeatures
      .join(labels, "name")
      .select($"label", $"features")


    /**
      * 将处理好的数据保存为SVM
      *
      */

    data
      .write
      .format("libsvm")
      .mode(SaveMode.Overwrite)
      .save("data/image")
  }

}
